package com.pro.common.service.websocket.service;

import cn.hutool.json.JSONUtil;
import com.pro.baoyi.common.api.BaoyiConst;
import com.pro.baoyi.common.api.enums.EnumRabbitMqRoutingKey;
import com.pro.common.api.model.ILoginInfo;
import com.pro.common.service.websocket.model.base.WebSocketMessage;
import com.pro.common.service.websocket.model.base.WebsocketClientToken;
import com.pro.common.service.websocket.mq.WebsocketMqProducer;
import com.pro.common.service.websocket.mq.model.BaseWebsocketNotifyData;
import com.pro.common.service.websocket.util.IPAddressUtil;
import com.pro.framework.api.util.JSONUtils;
import com.pro.framework.api.util.StrUtils;
import com.pro.framework.cache.CacheManagerCenterRedis;
import com.pro.framework.cache.CombinedCacheManager;
import org.jetbrains.annotations.NotNull;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.cache.Cache;
import org.springframework.cache.support.SimpleValueWrapper;
import org.springframework.stereotype.Service;
import org.springframework.web.socket.TextMessage;
import org.springframework.web.socket.WebSocketSession;

import java.io.IOException;
import java.time.LocalDateTime;
import java.util.*;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

@Service
public class WebsocketEventService {
    @Autowired
    private WebsocketMqProducer websocketMqProducer;

    @Autowired
    private CacheManagerCenterRedis cacheManagerCenterRedis;

    @Autowired
    private CombinedCacheManager combinedCacheManager;

    @Autowired
    private IPAddressUtil ipAddressUtil;

    public WebSocketMessage getWebSocketMessage(Long userId, ILoginInfo loginInfo, String connId, String tokenMd5) {
        boolean canLogin;
        Cache cache = combinedCacheManager.getCache(BaoyiConst.CacheKey.LOGIN_LIMIT_TOKEN);
        Object value = cache.get(userId + "_" + loginInfo.getLoginEntrance());
        List<WebsocketClientToken> clients = null;
        if (value instanceof SimpleValueWrapper) {
            Object wrappedValue = ((SimpleValueWrapper) value).get();
            if (wrappedValue instanceof List) {
                clients = (List<WebsocketClientToken>) wrappedValue;
            }
        }
        WebSocketMessage websocketMessage = new WebSocketMessage();
        websocketMessage.setConnId(connId);
        if (clients == null||clients.isEmpty()) {
            clients = new ArrayList<>();
            cache.put(userId + "_" + loginInfo.getLoginEntrance(), clients); // 记得把新列表放回缓存
        }
        Map<String,WebsocketClientToken> clientsMap = clients.stream().collect(Collectors.toMap(WebsocketClientToken::getClientId, c -> c));
        Set<String> tokenSet = clients.stream().map(WebsocketClientToken::getToken).collect(Collectors.toSet());
        if (clientsMap.containsKey(connId)) {
            WebsocketClientToken oldClient = clientsMap.get(connId);
            if (oldClient != null) {
                WebsocketClientToken websocketClientToken = clientsMap.get(connId);
                websocketClientToken.setLastActiveTime(LocalDateTime.now());
            }
            canLogin = true;
        }else{
            int loginMaxNum = 0;
            Integer obj = cacheManagerCenterRedis.get(BaoyiConst.CacheKey.LOGIN_LIMIT_DEVICE, userId + "_" + loginInfo.getLoginEntrance(), Integer.class);
            if (obj != null) {
                loginMaxNum = obj;
            }
            if(loginMaxNum > clients.stream().map(WebsocketClientToken::getToken).distinct().count()) {
                clients.add(new WebsocketClientToken(userId, loginInfo.getLoginEntrance(), tokenMd5, connId, LocalDateTime.now()));
                canLogin = true;
            } else if (tokenSet.contains(tokenMd5)) {
                clients.add(new WebsocketClientToken(userId, loginInfo.getLoginEntrance(), tokenMd5, connId, LocalDateTime.now()));
                canLogin = true;
            } else{
                canLogin = false;
            }
        }
        if(canLogin){
            websocketMessage.setType("LOGIN_SUCCESS");
            websocketMessage.setCode(200);
            websocketMessage.setMessage("登录成功");
        }else{
            websocketMessage.setType("LOGIN_LIMIT");
            websocketMessage.setCode(409);
            websocketMessage.setMessage("登录设备已达上限");
            websocketMessage.setData(clients);
        }
        cache.put(userId +"_"+ loginInfo.getLoginEntrance(), clients);
        cacheManagerCenterRedis.put(BaoyiConst.CacheKey.LOGIN_LIMIT_USER, userId +"_"+ loginInfo.getLoginEntrance()+"_"+ipAddressUtil.getServerAddress(), clients.size());
        cacheManagerCenterRedis.put(BaoyiConst.CacheKey.LOGIN_LIMIT_TOKEN_INDEX, tokenMd5, userId +"_"+ loginInfo.getLoginEntrance(),14L, TimeUnit.DAYS);
        return websocketMessage;
    }

    public WebSocketMessage handleMessage(String connId,String token,ILoginInfo loginInfo,WebSocketSession session, TextMessage message) throws IOException {
        String payload = message.getPayload();
        BaseWebsocketNotifyData wsMessage = payload.isEmpty() ? new BaseWebsocketNotifyData() : JSONUtil.toBean(payload, BaseWebsocketNotifyData.class);
        Cache cache = combinedCacheManager.getCache(BaoyiConst.CacheKey.LOGIN_LIMIT_TOKEN);
        Long userId = loginInfo.getId();
        Object value = cache.get(userId + "_" + loginInfo.getLoginEntrance());
        List<WebsocketClientToken> clients = null;
        if (value instanceof SimpleValueWrapper) {
            Object wrappedValue = ((SimpleValueWrapper) value).get();
            if (wrappedValue instanceof List) {
                clients = (List<WebsocketClientToken>) wrappedValue;
            }
        }
        WebSocketMessage webSocketMessage = new WebSocketMessage();
        switch (wsMessage.getBizType()){
            case KEEP_ALIVE -> {
                webSocketMessage = new WebSocketMessage("KEEP_ALIVE", 200, "连接成功", connId, null);
                if (token != null && connId != null) {
                    // 本地记录
                    String tokenMd5 = StrUtils.md5(token);
                    Map<String,WebsocketClientToken> clientsMap = clients.stream().collect(Collectors.toMap(WebsocketClientToken::getClientId, c -> c));
                    long count = clients.stream().map(WebsocketClientToken::getClientId).distinct().count();
                    if (clientsMap.containsKey(connId)) {
                        WebsocketClientToken oldClient = clientsMap.get(connId);
                        if (oldClient != null) {
                            oldClient.setLastActiveTime(LocalDateTime.now());
                        }
                    }else{

                        int loginMaxNum = 0;
//                        Object obj = cacheManagerCenterRedis.get(BaoyiConst.CacheKey.LOGIN_LIMIT_DEVICE, userId + "_" + loginInfo.getLoginEntrance());
//                        if (obj != null) {
//                            loginMaxNum = ((Number) obj).intValue();
//                        }
                        Integer obj = cacheManagerCenterRedis.get(BaoyiConst.CacheKey.LOGIN_LIMIT_DEVICE, userId + "_" + loginInfo.getLoginEntrance(), Integer.class);
                        if (obj != null) {
                            loginMaxNum = obj;
                        }
                        if(loginMaxNum > count) {
                            clients.add(new WebsocketClientToken(userId, loginInfo.getLoginEntrance(), tokenMd5, connId, LocalDateTime.now()));
                            cache.put(userId +"_"+loginInfo.getLoginEntrance(), clients);
                            cacheManagerCenterRedis.put(BaoyiConst.CacheKey.LOGIN_LIMIT_TOKEN_INDEX, tokenMd5, userId +"_"+ loginInfo.getLoginEntrance(),14L, TimeUnit.DAYS);
                        }
                    }
                    cacheManagerCenterRedis.put(BaoyiConst.CacheKey.LOGIN_LIMIT_USER, userId +"_"+loginInfo.getLoginEntrance()+"_"+ipAddressUtil.getServerAddress(), count);
                }
            }case GET_LOGIN_LIST -> {
                webSocketMessage = new WebSocketMessage("GET_LOGIN_LIST", 200, "获取登录列表成功", connId, clients.stream().collect(Collectors.collectingAndThen(
                        Collectors.toCollection(() ->
                                new TreeSet<>(Comparator.comparing(WebsocketClientToken::getToken))
                        ),
                        ArrayList::new
                )));
            }default -> {
                websocketMqProducer.sendWebsocketEvent(EnumRabbitMqRoutingKey.websocket_sync, message.getPayload());
            }
        }
        return webSocketMessage;
    }

    public void closeClean(Long userId, ILoginInfo loginInfo, String connId, String tokenMd5) {
        if (userId != null) {
            Cache cache = combinedCacheManager.getCache(BaoyiConst.CacheKey.LOGIN_LIMIT_TOKEN);
            Object value = cache.get(userId + "_" + loginInfo.getLoginEntrance());
            List<WebsocketClientToken> clients = null;

            if (value instanceof SimpleValueWrapper) {
                Object wrappedValue = ((SimpleValueWrapper) value).get();
                if (wrappedValue instanceof List) {
                    clients = (List<WebsocketClientToken>) wrappedValue;
                }
            }
            if(clients!=null) {
                Map<String, WebsocketClientToken> clientsMap = clients.stream().collect(Collectors.toMap(WebsocketClientToken::getClientId, c -> c));
                long count = clients.stream().map(WebsocketClientToken::getClientId).distinct().count();
                if (clientsMap.containsKey(connId)) {
                    clients.remove(clientsMap.get(connId));
                }
                cache.put(userId + "_" + loginInfo.getLoginEntrance(), clients);
                Set<String> tokenSet = clients.stream()
                        .map(WebsocketClientToken::getToken)
                        .collect(Collectors.toSet());
                boolean exists = tokenSet.contains(tokenMd5);
                if (!exists) {
                    cacheManagerCenterRedis.put(BaoyiConst.CacheKey.LOGIN_LIMIT_USER, userId + "_" + loginInfo.getLoginEntrance() + "_" + ipAddressUtil.getServerAddress(), count);
                }
            }
        }
    }
    public void deleteCacheByToken(String tokenMd5) {
        // 删除 token
        Cache cache = combinedCacheManager.getCache(BaoyiConst.CacheKey.LOGIN_LIMIT_TOKEN);
        String tokenKey = (String) cacheManagerCenterRedis.get(BaoyiConst.CacheKey.LOGIN_LIMIT_TOKEN_INDEX, tokenMd5, String.class);
        Object value = cache.get(tokenKey);
        List<WebsocketClientToken> clients = null;
        if (value instanceof SimpleValueWrapper) {
            Object wrappedValue = ((SimpleValueWrapper) value).get();
            if (wrappedValue instanceof List) {
                clients = (List<WebsocketClientToken>) wrappedValue;
            }
        }
        if (clients != null) {
            clients = clients.stream()
                    .filter(client -> !tokenMd5.equals(client.getToken()))
                    .collect(Collectors.toList());
            cache.put(tokenKey, clients);
            cacheManagerCenterRedis.decrement(BaoyiConst.CacheKey.LOGIN_LIMIT_USER, tokenKey + ipAddressUtil.getServerAddress(), 1);
        }
    }
}
